from pyspark import SparkContext, SparkConf
import csv, StringIO

conf = SparkConf().setMaster("spark://10.2.3.41:7077").setAppName("SPK_file")
sc = SparkContext.getOrCreate(conf)
sc.setLogLevel("ERROR")


# 将整个CSV文本文件转换成字典列表
def mapToDict(f):
    _, text = f
    input = StringIO.StringIO(text)
    csv_reader = csv.DictReader(input)
    users = []
    for line in csv_reader:
        users.append(line)
    return users


csvRDD = sc.wholeTextFiles("file:///media/psf/Home/Workspace/Rimi/P1901/lessons/spark/users.csv").flatMap(mapToDict)

# 将RDD所有的元素合并成一个列表
def combine_func(item, item1):
    if isinstance(item, list):
        item.append(item1)
        return item
    else:
        return [item, item1]

# 将列表中的字典输出到CSV文件
def write_to_csv(records):
    with open("./users2.csv", "w") as fp:
        csv_writer = csv.DictWriter(fp, fieldnames=("age", "name", "address"))
        csv_writer.writeheader()
        for x in records:
            csv_writer.writerow(x)


write_to_csv(csvRDD.reduce(f=combine_func))

# def map_func(x):
#     print(x)
#     words = re.split("\\s*,\\s*", x)
#     return words
#
#
# wordRDD = userRDD.flatMap(f=map_func)
# pairRDD = wordRDD.groupBy(lambda x: x)
# pairRDD1 = pairRDD.mapValues(lambda v: len([x for x in v]))
#
# pairRDD1.saveAsTextFile("file:///media/psf/Home/Workspace/Rimi/P1901/lessons/spark/result.txt")

# logsRDD = sc.wholeTextFiles("file:///media/psf/Home/Workspace/Rimi/P1901/lessons/spark/logs")
#
#
# def get_fruits(file):
#     pattern = "INFO:\\s+\\w+\\s+(?P<fruit>\\w+)"
#     file_name, value = file
#     name = os.path.basename(file_name).split(".")[0]
#
#     for line in value.split("\n"):
#         m = re.match(pattern, line)
#         if not m:
#             print(value)
#             continue
#         yield (name, m.group("fruit"))
#
#
# fruitRDD = logsRDD.flatMap(f=get_fruits)
# newRdd = fruitRDD.groupBy(lambda x: x)
# resultRdd = newRdd.mapValues(lambda x: len([w for w in x]))
# for w in resultRdd.collect():
#     print(w)

# fruitRDD = fruitRDD.groupByKey()

"""
[
('orange', 1),
('apple', 1),
('apple', 1),
('banana', 1)
]
"""

# def reduce_func(w):
#     total = {}
#     max_fruit = None
#     for f, c in w:
#         if max_fruit is None:
#             max_fruit = (f, c)
#         if f not in total:
#             total[f] = c
#         else:
#             total[f] += c
#
#         if total[f] > max_fruit[1]:
#             max_fruit = (f, total[f])
#
#     return max_fruit
#
# for w in fruitRDD.mapValues(reduce_func).collect():
#     print(w)


# for w in fruitRDD.collect():
#     print(w)


# def map_func2(x):
#     y = []
#     for f in x[1]:
#         y.append((f, x[0]))
#
#     return y
#     # return (x[1], x[0])
#
# rdd2 = rdd1.flatMap(map_func2)
# for w in rdd2.collect():
#     print(w)

sc.stop()
